/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server;

import java.io.Flushable;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.apache.zookeeper.server.paxos2.*;
import org.apache.zookeeper.proto.SetDataRequest;
import org.apache.zookeeper.proto.PrepareDataRequest;
import org.apache.zookeeper.proto.ReplyHeader;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.txn.SetDataTxn;
import org.apache.zookeeper.ZooDefs;

/**
 * This RequestProcessor logs requests to disk. It batches the requests to do
 * the io efficiently. The request is not passed to the next RequestProcessor
 * until its log has been synced to disk.
 */
public class SyncRequestProcessor extends Thread implements RequestProcessor {
    private static final Logger LOG = Logger
            .getLogger(SyncRequestProcessor.class);
    private final ZooKeeperServer zks;
    private final LinkedBlockingQueue<Request[]> queuedRequests = new LinkedBlockingQueue<Request[]>();
    private final RequestProcessor nextProcessor;

    private Thread snapInProcess = null;
    private FlushThread flushThread = new FlushThread();
    private PrepareAckThread prepareAckThread = new PrepareAckThread();

    /**
     * Transactions that have been written and are waiting to be flushed to
     * disk. Basically this is the list of SyncItems whose callbacks will be
     * invoked after flush returns successfully.
     */
    private LinkedList<Request[]> toFlush = new LinkedList<Request[]>();
    private final Random r = new Random(System.nanoTime());
    /**
     * The number of log entries to log before starting a snapshot
     */
    private static int snapCount = ZooKeeperServer.getSnapCount();

    private final Request requestOfDeath = Request.requestOfDeath;

    public SyncRequestProcessor(ZooKeeperServer zks,
                                RequestProcessor nextProcessor) {
        super("SyncThread:" + zks.getServerId());
        this.zks = zks;
        this.nextProcessor = nextProcessor;
        //flushThread.start();
        prepareAckThread.start();
    }

    /**
     * used by tests to check for changing snapcounts
     *
     * @param count
     */
    public static void setSnapCount(int count) {
        snapCount = count;
    }

    /**
     * used by tests to get the snapcount
     *
     * @return the snapcount
     */
    public static int getSnapCount() {
        return snapCount;
    }

    private void printQueue(){
	System.out.println("sync queue size="+queuedRequests.size());
	int prepareCount=0;
	int setCount = 0;
	for(Request[] reqs:queuedRequests){
	    if(reqs[0].type==OpCode.prepareData)
		prepareCount++;
	    else if(reqs[0].type==OpCode.setData)
		setCount++;
	    else
		System.out.println("unknown "+reqs[0].type);
	}
	System.out.println("prepareCount="+prepareCount+" setCount="+setCount);
	
    }

    @Override
    public void run() {
        try {
            int logCount = 0;

            // we do this in an attempt to ensure that not all of the servers
            // in the ensemble take a snapshot at the same time
            int randRoll = r.nextInt(snapCount / 2);
            long count = 0;
            long totalTime = 0;
            long waitTime = 0;
            long writeTime = 0;
            long otherTime = 0;
            int prepareCount = 0;
	    long reqCount = 0;
            long loopStart = 0;
            while (true) {

                if (count >= 4)
                    loopStart = System.currentTimeMillis();
                long start = System.currentTimeMillis();
                Request[] si = null;
                if (toFlush.isEmpty()) {
                    si = queuedRequests.take();
                } else {
                    si = queuedRequests.poll();//5, java.util.concurrent.TimeUnit.MILLISECONDS);
		    //if(queuedRequests.size()>2000)
		    //    System.out.println("syncSize="+queuedRequests.size());
                    if (si == null) {
			if (count > 4)
                    	    waitTime += (System.currentTimeMillis() - start);
                	start = System.currentTimeMillis();

                        flush(toFlush);
                        //flushThread.beginFlush(toFlush);
                        toFlush = new LinkedList<Request[]>();
                        prepareCount = 0;
                        if (count > 4) {
                            otherTime += (System.currentTimeMillis() - start);
                            totalTime += (System.currentTimeMillis() - loopStart);
                        }
                        continue;
                    }
                }
                if (si[0] == requestOfDeath) {
                    break;
                }
                count++;
                if (count > 4)
                    waitTime += (System.currentTimeMillis() - start);
                start = System.currentTimeMillis();
                /*
                     * if(si.request == null) System.out.println("request is null");
                     * if(si.txn == null) System.out.println("txn is null");
                     * if(si.hdr == null) System.out.println("hdr is null");
                     */
                /*if (si.type == OpCode.setData) {
                        SetDataTxn txn = (SetDataTxn) (si.txn);

                        PrepareDataRequest tmp = zks.prepareDataCache.get(txn.getSessionId(), si.cxid);
                        if(tmp != null){
                            LOG.info("Find matching in cache " + txn.getSessionId()+" "+txn.getBlockId());
                            txn.setData(tmp.getData());
                        }
                        else{
                            LOG.info("Do not find matching in cache " + txn.getSessionId()+" "+txn.getBlockId());
                            si.txn = null;
                        }
                    }*/
                if (si != null) {
                    // track the number of records written to the log
                    if (zks.getZKDatabase().append(si)) {
                        if (count > 4)
                            writeTime += (System.currentTimeMillis() - start);
                        start = System.currentTimeMillis();
			if (si[0].type == OpCode.prepareData)
                            logCount += si.length;
                        if (logCount > (snapCount / 2 + randRoll)) {
                            randRoll = r.nextInt(snapCount / 2);
                            // roll the log
                            zks.getZKDatabase().rollLog();
                            // take a snapshot
                            if (Configuration.isInRecovery()||Configuration.isBusy()||(snapInProcess != null
                                    && snapInProcess.isAlive())) {
                                LOG.warn("Too busy to snap, skipping");
                            } else {
                                snapInProcess = new Thread("Snapshot Thread") {
                                    public void run() {
                                        try {
                                            zks.takeSnapshot();
                                        } catch (Exception e) {
                                            LOG.warn("Unexpected exception", e);
                                        }
                                    }
                                };
                                snapInProcess.start();
                            }
                            logCount = 0;
                        }
                    } else if (toFlush.isEmpty()) {
                        // optimization for read heavy workloads
                        // iff this is a read, and there are no pending
                        // flushes (writes), then just pass this to the next
                        // processor
                        nextProcessor.processRequest(si);
                        if (nextProcessor instanceof Flushable) {
                            ((Flushable) nextProcessor).flush();
                        }
                        continue;
                    }

                    toFlush.add(si);
                    if (si[0].type == OpCode.prepareData){
                        prepareCount += si.length;
			reqCount += si.length;
		    }
                    if (prepareCount >= Configuration.syncInterval) {
                        flush(toFlush);
                        //flushThread.beginFlush(toFlush);
                        toFlush = new LinkedList<Request[]>();
                        prepareCount = 0;
                    }
                    if (count > 4)
                        otherTime += (System.currentTimeMillis() - start);
		    //System.out.println("other time ="+otherTime);
                    start = System.currentTimeMillis();

                }
		/*if(Runtime.getRuntime().freeMemory()<Runtime.getRuntime().maxMemory()/2){
                    System.out.println("gc");
                    System.gc();
                }*/

                if (count > 4)
                    totalTime += (System.currentTimeMillis() - loopStart);
                if (count % 1000 == 999)
                    System.out.println("wait=" + (double) waitTime / (double) totalTime + " write=" + (double) writeTime / (double) totalTime + " other=" + (double) otherTime / (double) totalTime + " reqCount=" + reqCount + " " + reqCount * 1000 / totalTime + " reqs/sec" + " totalTime=" + totalTime);
            }
        } catch (Throwable t) {
            LOG.fatal("Severe unrecoverable error, exiting", t);
            System.exit(11);
        }
        LOG.info("SyncRequestProcessor exited!");
    }

    int flushCount = 0;
    long flushTime = 0;

    private void flush(LinkedList<Request[]> toFlush) throws IOException {
        if (toFlush.isEmpty())
            return;
        long start = System.currentTimeMillis();
	//System.out.println("flush "+toFlush.size());
        zks.getZKDatabase().commit();
	long time = System.currentTimeMillis()-start;
	if(time>3000){
	    System.out.println("long sync time "+time);
	    Runtime.getRuntime().gc();
	}
        prepareAckThread.addPrepare(toFlush);
	//flushTime += (System.currentTimeMillis() - start);
	//System.out.println("flush time="+flushTime);
	//flushCount++;
        //System.out.println("flush time " + (double)(flushTime)/(double)flushCount);
        /*while (!toFlush.isEmpty()) {
            Request[] si = toFlush.remove();
            if (si[0].type == ZooDefs.OpCode.prepareData) {
                flushCount+=si.length;
                prepareAckThread.addPrepare(si);
            } else {
                nextProcessor.processRequest(si);
            }
        }
        if (nextProcessor instanceof Flushable) {
            ((Flushable) nextProcessor).flush();
        }
        flushTime += (System.currentTimeMillis() - start);*/
    }

    public void shutdown() {
        queuedRequests.add(new Request[]{requestOfDeath});
        try {
            this.join();
        } catch (InterruptedException e) {
            LOG.warn("Interrupted while wating for " + this + " to finish");
        }
        nextProcessor.shutdown();
    }

    public void processRequest(Request[] request) {
//	if(queuedRequests.size()>100)
//	    printQueue();
        // request.addRQRec(">sync");
        queuedRequests.add(request);
    }

    public void processRequest(Request request) {
        throw new RuntimeException("Not implemented");
    }

    private class PrepareAckThread extends Thread {
        private LinkedBlockingQueue<LinkedList<Request[]>> prepareAcks = new LinkedBlockingQueue<LinkedList<Request[]>>();

        public void addPrepare(LinkedList<Request[]> toFlush) {
            try {
                prepareAcks.put(toFlush);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public void run() {
            while (true) {
                try {
		    LinkedList<Request[]> toFlush = prepareAcks.take();
		    while (!toFlush.isEmpty()) {
            		Request[] si = toFlush.remove();
            		if (si[0].type == ZooDefs.OpCode.prepareData) {
                            for (int i = 0; i < si.length; i++) {
                            	zks.decInProcess();
                            	ReplyHeader hdr = new ReplyHeader(si[i].cxid, si[i].nonce, si[i].type, -1, 0);

                            	ServerCnxn cnxn = zks.getServerCnxnFactory().getServerCnxn(si[i].hdr.getClientId());
                            	try {
                                    if(cnxn!=null)
                                    	cnxn.sendResponse(hdr, null, "response");
                            	}
                            	catch (IOException e) {
                                    LOG.error("Cannot send reply fro prepare ", e);
                            	}
                            }
            		} else {
                	    nextProcessor.processRequest(si);
            		}
        	    }
        	    if (nextProcessor instanceof Flushable) {
            		((Flushable) nextProcessor).flush();
        	    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

        }
    }


    private class FlushThread extends Thread {
        private LinkedBlockingQueue<LinkedList<Request[]>> flushReqs = new LinkedBlockingQueue<LinkedList<Request[]>>();

        public void beginFlush(LinkedList<Request[]> toFlush) {
            try {
                flushReqs.put(toFlush);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public void run() {
            while (true) {
                try {
                    LinkedList<Request[]> toFlush = flushReqs.take();
                    flush(toFlush);
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
                catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


    }

}
